package cn.jly.flink.state;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 所谓的状态指的是，在流处理过程中那些需要记住的数据，而这些数据既可以包括业务数据，也可以包括元数据。Flink 本身提供了不同的状态管理器来管理状态，并且这个状态可以非常大。
 * <p>
 * Flink 的状态数据可以存在 JVM 的堆内存或者堆外内存中，当然也可以借助第三方存储，例如 Flink 已经实现的对 RocksDB 支持。
 * <p>
 * Flink 的官网同样给出了适用于状态计算的几种情况: 复杂事件处理获取符合某一特定时间规则的事件、聚合计算、机器学习的模型训练、使用历史的数据进行计算。
 *
 * @author lanyangji
 * @create 2020-09-03 15:42
 */
public class StateApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 指定状态后端，FsStateBackend 因为将状态存储在了外部系统如 HDFS 中，所以它适用于大作业、状态较大、全局高可用的那些任务
        env.setStateBackend(new FsStateBackend("file:///e:/checkpoint", false));

        env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 5L), Tuple2.of(1L, 2L))
                .keyBy((KeySelector<Tuple2<Long, Long>, Long>) value -> value.f0)
                .flatMap(new CountWindowAverage())
                .printToErr();

        env.execute("StateApp");
    }

    public static class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
        private transient ValueState<Tuple2<Long, Long>> sum;

        @Override
        public void open(Configuration parameters) throws Exception {
            ValueStateDescriptor<Tuple2<Long, Long>> describe = new ValueStateDescriptor<>(
                    "average",
                    TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {
                    })
            );

            // ttl(time to live)，存活时间（过期时间）配置
            // UpdateType 表明了过期时间什么时候更新，而对于那些过期的状态，是否还能被访问则取决于 StateVisibility 的配置
            StateTtlConfig stateTtlConfig = StateTtlConfig
                    .newBuilder(Time.seconds(10))
                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                    .build();

            describe.enableTimeToLive(stateTtlConfig);

            sum = getRuntimeContext().getState(describe);
        }

        @Override
        public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
            Tuple2<Long, Long> currentSum;
            // 访问valueState
            if (null == sum.value()) {
                currentSum = Tuple2.of(0L, 0L);
            } else {
                currentSum = sum.value();
            }

            // 更新
            currentSum.f0 += 1;
            currentSum.f1 += value.f1;

            // 更新state
            sum.update(currentSum);

            // 如果count的值大于等于2，求值并清空state
            if (currentSum.f0 >= 2) {
                out.collect(Tuple2.of(value.f0, currentSum.f1 / currentSum.f0));
                sum.clear();
            }
        }
    }
}
